package com.dec.kks.etl;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.storage.StorageLevel;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.net.InetAddress;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

public class TCPMain {

    public static void main(String[] args) throws Exception {
        System.setProperty("hadoop.home.dir","/home/hdfs/bigdata/hadoop-2.7.4");
        String hostname = InetAddress.getLocalHost().getHostName();
        System.out.println(hostname);
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("word count with socket source");

        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));
        JavaReceiverInputDStream<Byte> is = jssc.socketStream(hostname, 8800, new Function<InputStream, Iterable<Byte>>() {
            @Override
            public Iterable<Byte> call(InputStream inputStream) throws Exception {
                ByteArrayOutputStream result = new ByteArrayOutputStream();
                byte[] buffer = new byte[1024];
                int length;
                while ((length = inputStream.read(buffer)) != -1) {
                    result.write(buffer, 0, length);
                    for (int i = 0; i <1024 ; i++) {
                        System.out.println("字节数据："+buffer[i]);
                    }
                }

                return null;
            }
        }, StorageLevel.MEMORY_ONLY());

        is.print();
        jssc.start();
        jssc.awaitTermination();
    }
}
